package com.nl.loginfaildetect;

import com.nl.bean.input.LoginLogEvent;
import com.nl.bean.output.Warning;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author shihb
 * @date 2019/12/24 16:02
 * 2秒内登录失败次数超过2次的简单实现
 */
public class LoginFailDetect {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.setParallelism(1);
    DataStreamSource<String> source = env.readTextFile(
        LoginFailDetect.class.getClassLoader().getResource("LoginLog.csv").getPath());
    SingleOutputStreamOperator<LoginLogEvent> eventStream = source
        .map(s -> {
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long loginTime = Long.parseLong(arr[3].trim());
          return new LoginLogEvent(userId, arr[1].trim(), arr[2].trim(), loginTime);
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginLogEvent>(
            Time.seconds(5)) {
          @Override
          public long extractTimestamp(LoginLogEvent element) {
            return element.getLoginTime() * 1000;
          }
        });

    SingleOutputStreamOperator<Warning> WarnStream = eventStream.keyBy(loginLogEvent->loginLogEvent.getUserId())
        .process(new LoginFailFunction(2));
    WarnStream.print();

    env.execute("login fail job");



  }

}
class LoginFailFunction extends KeyedProcessFunction <Long,LoginLogEvent,Warning>{
  private int loginFailCap;
  private ListState<LoginLogEvent> failTimes;

  public LoginFailFunction(int i) {
    this.loginFailCap=i;
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    failTimes=getRuntimeContext().getListState(new ListStateDescriptor<LoginLogEvent>("failTimes-state",LoginLogEvent.class));

  }

  @Override
  public void processElement(LoginLogEvent value, Context ctx, Collector<Warning> out)
      throws Exception {

    String curLoginState=value.getLoginState().toLowerCase();
    // 方法一:设置一个2秒后的定时器，根据失败的次数来告警
    // 缺陷就是2秒后才触发，2秒内可能产生n次的登录失败
//    if("fail".equals(curLoginState) ){
//      //登录失败并且第一次就注册定时器
//      if(!failTimes.get().iterator().hasNext()){
//        ctx.timerService().registerEventTimeTimer(value.getLoginTime()*1000+2000);
//      }
//      this.failTimes.add(value);
//    }else if ("success".equals(curLoginState)){
//      //登录成功清空状态
//      failTimes.clear();
//    }
    // 方法二:来一条就判断并告警
    // 缺陷如果2秒内失败n次就很难实现,乱序数据处理很复杂
    if("fail".equals(curLoginState) ){
      //登录失败并且之前有失败就判断
      if(failTimes.get().iterator().hasNext()){
        long lastTime=failTimes.get().iterator().next().getLoginTime()*1000;
        long now=value.getLoginTime()*1000;
        if(now - lastTime<= 2000){
          // 两次间隔小于2秒，输出报警信息
          out.collect(new Warning(value.getUserId(),new Timestamp(lastTime).toString(),new Timestamp(now).toString(), "2秒失败了2次"));
        }
      }
      // 更新最近一次的失败事件
      this.failTimes.clear();
      this.failTimes.add(value);
    }else if ("success".equals(curLoginState)){
      //登录成功清空状态
      failTimes.clear();
    }
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<Warning> out) throws Exception {
    //触发定时器的时候，根据状态的个数决定是否触发告警
    ArrayList<LoginLogEvent> failList = new ArrayList<LoginLogEvent>();
    Iterator<LoginLogEvent> iter = failTimes.get().iterator();
    while (iter.hasNext()){
      failList.add(iter.next());
    }
    if(failList.size()>=this.loginFailCap){
      LoginLogEvent headElement = failList.get(0);
      LoginLogEvent lastElement = failList.get(failList.size()-1);
      out.collect(new Warning(headElement.getUserId(),new Timestamp(headElement.getLoginTime()*1000).toString(),new Timestamp(lastElement.getLoginTime()*1000).toString(), "2秒失败了"+failList.size()+"次"));
      this.failTimes.clear();
    }
  }
}

